package app_kvServer;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.log4j.Logger;

import common.messages.KVAdminMessage;
import common.messages.KVMessageImpl;
import common.messages.MD5Hash;
import common.messages.MetaData;
import communicationLogic.ClientSocket;

/**
 * @author Zhou, Shi
 * 
 */
public class ClientConnection implements Runnable {
	private static Logger logger = Logger.getRootLogger();

	private boolean isOpen = true;
	private static final int BUFFER_SIZE = 1024;
	private static final int DROP_SIZE = 128 * BUFFER_SIZE;
	private static final int KEY_LENGTH = 20;
	private static final int VALUE_LENGTH = 120000;
	private static final String TAIL = "ffffffffffffffffffffffffffffffff";
	private static final String HEAD = "00000000000000000000000000000000";
	private static final int CLIENT_TAG = 49;// 1
	private static final int ECS_TAG = 50; // 2
	private static final int SERVER_TAG = 51; // 3
	private static final char LINE_FEED = 0x0A;
	private static final char RETURN = 0x0D;

	private Socket clientSocket;
	private InputStream input;
	private OutputStream output;
	private int port;

	private Singleton instance;

	/**
	 * @param client
	 *            an client socket connection
	 * @param map
	 *            a hash map
	 */
	public ClientConnection(Socket client) {
		this.clientSocket = client;
		this.port = client.getLocalPort();
		this.isOpen = true;
		this.instance = Singleton.getInstance();
	}

	@Override
	public void run() {
		try {
			output = clientSocket.getOutputStream();
			input = clientSocket.getInputStream();

			// sendMessage(new
			// KVMessageImpl("Connection to MSRG Echo server established: "
			// +clientSocket.getLocalAddress().toString()+" "
			// + clientSocket.getLocalPort(),"" ,"GET_ERROR"));

			while (isOpen) {
				//try {
					byte[] receive = receiveMessage();
					switch (receive[0]) {
					case ECS_TAG:
						KVAdminMessage kvadmin = KVAdminMessage
								.toKVAdminMessage(receive);
						if (kvadmin != null) {
						    logger.info("command is : "
                                    + kvadmin.getCommand());
							logger.info("command is : "
									+ kvadmin.getCommand());
							logger.info("MetaDatalength is :  "
                                    + kvadmin.getMetaList().size());
							logger.info("MetaDatalength is :  "
									+ kvadmin.getMetaList().size());
							if (kvadmin.getCommand().equalsIgnoreCase("init")) {
								updateMetaData(kvadmin.getMetaList());
								// TODO send ack back
							} else if (kvadmin.getCommand().equalsIgnoreCase(
									"start")) {
								instance.setStarted(true);
								sendMessage("ACK".getBytes());

							} else if (kvadmin.getCommand().equalsIgnoreCase(
									"stop")) {
								instance.setStarted(false);
							} else if (kvadmin.getCommand().equalsIgnoreCase(
									"shutdown")) {
								System.exit(0);
							}else if (kvadmin.getCommand().equalsIgnoreCase(
                                    "update")) {
	                             updateMetaData(kvadmin.getMetaList());
                            } else if (kvadmin.getCommand().equalsIgnoreCase(
									"addNode")) {
								instance.setWriteLock(true);

								MetaData metaData = new MetaData();
                                for (MetaData newM : kvadmin.getMetaList()) {
                                    for (MetaData oldM : instance.getmList()) {
                                       String NewipPort = newM.getIp()+newM.getPort();
                                       String OldipPort = oldM.getIp()+oldM.getPort();
                                        if(!NewipPort.equalsIgnoreCase(OldipPort)){
                                            metaData = newM;
                                        }
                                        
                                    }
                                }
//								ArrayList<String> newipList = new ArrayList<String>();
//								ArrayList<String> oldipList = new ArrayList<String>();
//								for (MetaData m : kvadmin.getMetaList()) {
//									newipList
//											.add(m.getIp() + ":" + m.getPort());
//								}
//								for (MetaData m : instance.getmList()) {
//									oldipList
//											.add(m.getIp() + ":" + m.getPort());
//								}
//								for (int i = 0; i < newipList.size(); i++) {
//									if (!oldipList.contains(newipList.get(i))) {
//										metaData = kvadmin.getMetaList().get(i);
//									}
//								}
								String[] moverange = { metaData.getHead(),
										metaData.getTail() };

								logger.info("new Server"
										+ metaData.getPort());
								moveData(moverange, metaData.getIp(),
										metaData.getPort());
								removeData(moverange);
								updateMetaData(kvadmin.getMetaList());
								instance.setWriteLock(false);
								instance.setStarted(true);
							} else if (kvadmin.getCommand().equalsIgnoreCase(
									"removeNode")) {
								instance.setWriteLock(true);
								MetaData metaData = new MetaData();
								//
								for (MetaData newM : kvadmin.getMetaList()) {
								    for (MetaData oldM : instance.getmList()) {
								       String NewipPort = newM.getIp()+newM.getPort();
	                                   String OldipPort = oldM.getIp()+oldM.getPort();
								        if(!NewipPort.equalsIgnoreCase(OldipPort)){
								            metaData = newM;
								        }
								        
								    }
								}
								
								String[] moverange = { metaData.getHead(),
										metaData.getTail() };
								moveData(moverange, metaData.getIp(),
										metaData.getPort());
								//removeData(moverange);
								updateMetaData(kvadmin.getMetaList());
								instance.setWriteLock(false);
							}
						} else {
						    logger.info("server received nothing from ECS");
						}
						break;
					case SERVER_TAG:
						// TODO receive and send server Message
					    logger.info(new String(receive));
						addData(receive);
						sendMessage("SUCCESS TO MOVE DATA".getBytes());
						logger.info("Success add data");

						break;
					case CLIENT_TAG:
						if (instance.isStarted()) {
							KVMessageImpl latestMsg = KVMessageImpl
									.toKVMessage(receive);
							if (latestMsg != null) {
								if (!instance.inRange(latestMsg.getKey(), port)) {
									KVMessageImpl reply = new KVMessageImpl(
											latestMsg.getKey(),
											"",
											"SERVER_NOT_RESPONSIBLE",
											metadata2String(instance.getmList()));
									sendMessage(reply.toByteArray(reply));
								} else {
									if (latestMsg.getStatus().toString()
											.contains("GET")) {
										String clientValue = getKV(latestMsg
												.getKey());
										if (clientValue.isEmpty()) {
											KVMessageImpl reply = new KVMessageImpl(
													latestMsg.getKey(),
													clientValue, "GET_ERROR",
													"");
											sendMessage(reply
													.toByteArray(reply));
										} else {
											KVMessageImpl reply = new KVMessageImpl(
													latestMsg.getKey(),
													clientValue, "GET_SUCCESS",
													"");
											sendMessage(reply
													.toByteArray(reply));
										}
									} else if (!instance.isWriteLock()) {
										if (latestMsg.getValue().toString()
												.equalsIgnoreCase("null")) {
											if (deleteKV(latestMsg.getKey())) {

												KVMessageImpl reply = new KVMessageImpl(
														latestMsg.getKey(),
														latestMsg.getValue(),
														"DELETE_SUCCESS", "");
												sendMessage(reply
														.toByteArray(reply));
											} else {
												KVMessageImpl reply = new KVMessageImpl(
														latestMsg.getKey(),
														latestMsg.getValue(),
														"DELETE_ERROR", "");
												sendMessage(reply
														.toByteArray(reply));
											}
										} else if (instance
												.getKvmap()
												.containsKey(
														new MD5Hash()
																.doHash(latestMsg
																		.getKey()))) {
											if (updateKV(latestMsg.getKey(),
													latestMsg.getValue())) {
												KVMessageImpl reply = new KVMessageImpl(
														latestMsg.getKey(),
														latestMsg.getValue(),
														"PUT_UPDATE", "");
												sendMessage(reply
														.toByteArray(reply));
											} else {
												KVMessageImpl reply = new KVMessageImpl(
														latestMsg.getKey(),
														latestMsg.getValue(),
														"PUT_ERROR", "");
												sendMessage(reply
														.toByteArray(reply));
											}
										} else {
											if (insertKV(latestMsg.getKey(),
													latestMsg.getValue())) {
												KVMessageImpl reply = new KVMessageImpl(
														latestMsg.getKey(),
														latestMsg.getValue(),
														"PUT_SUCCESS", "");
												System.out
														.println(new MD5Hash()
																.doHash(latestMsg
																		.getKey()));
												sendMessage(reply
														.toByteArray(reply));
											} else {
												KVMessageImpl reply = new KVMessageImpl(
														latestMsg.getKey(),
														latestMsg.getValue(),
														"PUT_ERROR", "");
												sendMessage(reply
														.toByteArray(reply));
											}
										}
									} else {
										KVMessageImpl reply = new KVMessageImpl(
												"", "", "SERVER_WRITE_LOCK", "");
										sendMessage(reply.toByteArray(reply));
									}
								}
							}
						} else {
							KVMessageImpl reply = new KVMessageImpl("", "",
									"SERVER_STOPPED", "");
							sendMessage(reply.toByteArray(reply));
						}
						break;
					default:
						break;
					}
				//} catch (IOException ioe) {
				//	logger.error("Error! Connection lost!");
				//	isOpen = false;
				//}
			}

		} catch (IOException ioe) {
			logger.error("Error! Connection could not be established!", ioe);

		} finally {
			try {
				if (clientSocket != null) {
					input.close();
					output.close();
					clientSocket.close();
				}
			} catch (IOException ioe) {
				logger.error("Error! Unable to tear down connection!", ioe);
			}
		}
	}

	/**
	 * @param msg
	 * @throws IOException
	 */
	public synchronized void sendMessage(byte[] msgBytes) throws IOException {
		output.write(msgBytes, 0, msgBytes.length);
		output.flush();
	}

	/**
	 * @return
	 * @throws IOException
	 */
	private synchronized byte[] receiveMessage() throws IOException {

		int index = 0;
		byte[] msgBytes = null, tmp = null;
		byte[] bufferBytes = new byte[BUFFER_SIZE];

		/* read first char from stream */
		byte read = (byte) input.read();
		boolean reading = true;

		while (read != 13 && reading) {/* carriage return */
			/* if buffer filled, copy to msg array */
			if (index == BUFFER_SIZE) {
				if (msgBytes == null) {
					tmp = new byte[BUFFER_SIZE];
					System.arraycopy(bufferBytes, 0, tmp, 0, BUFFER_SIZE);
				} else {
					tmp = new byte[msgBytes.length + BUFFER_SIZE];
					System.arraycopy(msgBytes, 0, tmp, 0, msgBytes.length);
					System.arraycopy(bufferBytes, 0, tmp, msgBytes.length,
							BUFFER_SIZE);
				}

				msgBytes = tmp;
				bufferBytes = new byte[BUFFER_SIZE];
				index = 0;
			}

			/* only read valid characters, i.e. letters and constants */
			bufferBytes[index] = read;
			index++;

			/* stop reading is DROP_SIZE is reached */
			if (msgBytes != null && msgBytes.length + index >= DROP_SIZE) {
				reading = false;
			}

			/* read next char from stream */
			read = (byte) input.read();
		}

		if (msgBytes == null) {
			tmp = new byte[index];
			System.arraycopy(bufferBytes, 0, tmp, 0, index);
		} else {
			tmp = new byte[msgBytes.length + index];
			System.arraycopy(msgBytes, 0, tmp, 0, msgBytes.length);
			System.arraycopy(bufferBytes, 0, tmp, msgBytes.length, index);
		}

		msgBytes = tmp;
		return msgBytes;
	}

	/**
	 * @param key
	 *            the new key
	 * @param value
	 *            the new value
	 * @return
	 */
	private synchronized boolean insertKV(String key, String value) {
		synchronized (this) {
			if (key.length() > KEY_LENGTH || value.length() > VALUE_LENGTH) {
				return false;
			} else {
				instance.getKvmap().put(new MD5Hash().doHash(key), value);
				return true;
			}
		}
	}

	/**
	 * @param key
	 *            the key ready to be updated
	 * @param value
	 *            the new value
	 * @return
	 */
	private synchronized boolean updateKV(String key, String value) {
		synchronized (this) {
			if (key.length() > KEY_LENGTH || value.length() > VALUE_LENGTH) {
				return false;
			} else {
				instance.getKvmap().remove(new MD5Hash().doHash(key));
				instance.getKvmap().put(new MD5Hash().doHash(key), value);
				return true;
			}
		}
	}

	/**
	 * @param key
	 * @return success or not
	 */
	private synchronized boolean deleteKV(String key) {
		synchronized (this) {
			if (instance.getKvmap().containsKey(new MD5Hash().doHash(key))) {
				instance.getKvmap().remove(new MD5Hash().doHash(key));
				return true;
			} else {
				return false;
			}
		}
	}

	/**
	 * @param key
	 * @return the key and value
	 */
	private String getKV(String key) {
		synchronized (this) {
			if (instance.getKvmap().containsKey(new MD5Hash().doHash(key))) {
				return instance.getKvmap().get(new MD5Hash().doHash(key))
						.toString();
			} else
				return "";
		}
	}

	private void updateMetaData(List<MetaData> mList) {
		instance.setmList(mList);
		for (MetaData m : mList) {
			if (m.getPort().equals(String.valueOf(port))) {
				instance.setStart_index(m.getHead());
				instance.setEnd_index(m.getTail());
			}
		}
	}

	private String metadata2String(List<MetaData> mList) {
		String meta = "";
		for (MetaData m : mList) {
			meta += m.getIp() + ":" + m.getPort() + ":" + m.getHead() + ":"
					+ m.getTail() + ";";
		}
		return meta;
	}

	//moveData(moverange, metaData.getIp(),metaData.getPort());
    //removeData(moverange);
	public boolean moveData(String[] range, String ip, String port) {
		// TODO using socket
		ClientSocket cs;
		try {
			cs = new ClientSocket(ip, Integer.parseInt(port));
			logger.info("IP:" + ip + "Port:" + port);
			cs.connect();
			
			cs.send(getData(range));
			logger.info("##send data is "+getData(range).toString());
			logger.info("Success send data");

			byte[] ctrBytes = new byte[] { LINE_FEED, RETURN };
			byte[] bytes = cs.receive();
			byte[] tmp = new byte[bytes.length + ctrBytes.length];
			System.arraycopy(bytes, 0, tmp, 0, bytes.length);
			System.arraycopy(ctrBytes, 0, tmp, bytes.length, ctrBytes.length);
			String answer = new String(tmp).trim();
			logger.info(answer);
			if (answer.contains("SUCCESS")) {
			    logger.info("Success move data");
				return true;
			} else {
				return false;
			}

		} catch (IOException e) {
			return false;
		} catch (Exception e) {
			return false;
		}
	}

	public byte[] getData(String[] dataRange) {
		String dataString = "3;";
		Iterator iter = instance.getKvmap().entrySet().iterator();
		while (iter.hasNext()) {
			Entry<String, String> entry = (Entry<String, String>) iter.next();
			String key = entry.getKey();
			if (dataRange[1].compareToIgnoreCase(dataRange[0]) > 0) {
				if (key.compareToIgnoreCase(dataRange[1]) < 0
						&& key.compareToIgnoreCase(dataRange[0]) > 0) {

					dataString += entry.getKey() + ":" + entry.getValue() + ";";
					logger.info(dataString);
				}
			} else {
				if ((key.compareToIgnoreCase(TAIL) < 0 && key
						.compareToIgnoreCase(dataRange[0]) > 0)
						|| (key.compareToIgnoreCase(dataRange[1]) < 0 && key
								.compareToIgnoreCase(HEAD) > 0)) {
					dataString += entry.getKey() + ":" + entry.getValue() + ";";
					logger.info(dataString);
				}
			}
		}
		byte[] bytes = dataString.getBytes();
		byte[] ctrBytes = new byte[] { LINE_FEED, RETURN };
		byte[] tmp = new byte[bytes.length + ctrBytes.length];

		System.arraycopy(bytes, 0, tmp, 0, bytes.length);
		System.arraycopy(ctrBytes, 0, tmp, bytes.length, ctrBytes.length);
		return tmp;
	}

	private void removeData(String[] dataRange) {

		HashMap<String, String> copy = (HashMap<String, String>) instance
				.getKvmap().clone();
		Iterator iter = copy.entrySet().iterator();
		while (iter.hasNext()) {
			Entry<String, String> entry = (Entry<String, String>) iter.next();
			String key = entry.getKey();
			if (dataRange[1].compareToIgnoreCase(dataRange[0]) > 0) {
				if (key.compareToIgnoreCase(dataRange[1]) < 0
						&& key.compareToIgnoreCase(dataRange[0]) > 0) {
					instance.getKvmap().remove(key);
					logger.info("Success remove data");
				}
			} else {
				if ((key.compareToIgnoreCase(TAIL) < 0 && key
						.compareToIgnoreCase(dataRange[0]) > 0)
						|| (key.compareToIgnoreCase(dataRange[1]) < 0 && key
								.compareToIgnoreCase(HEAD) > 0)) {
					instance.getKvmap().remove(key);
					logger.info("Success remove data");
				}
			}
		}
	}

	private void addData(byte[] data) {
		byte[] ctrBytes = new byte[] { LINE_FEED, RETURN };
		byte[] tmp = new byte[data.length + ctrBytes.length];
		System.arraycopy(data, 0, tmp, 0, data.length);
		System.arraycopy(ctrBytes, 0, tmp, data.length, ctrBytes.length);
		String str = new String(tmp).trim();
		String[] newData = str.split(";");
		for (int i = 1; i < newData.length; i++) {
			String[] pair = newData[i].split(":");
			instance.getKvmap().put(pair[0], pair[1]);
			logger.info("I have take over key-value:" +pair[0]+":"+pair[1]);
		}
	}
}
